
/***************************************************************************
 *   sakcab 
 *                                                                         *
 *   Copyright (C) 2010 by kuerant                                         *
 *                         <kuerant@gmail.com>                             *
 *                                                                         *
 *   This library is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU Library General Public License as       *
 *   published by the Free Software Foundation; either version 2 of the    *
 *   License, or (at your option) any later version.                       *
 *                                                                         *
 *   This library is distributed in the hope that it will be usefull, but  *
 *   WITHOUT ANY WARRANTY; without even the implied warranty of            *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
 *   Library General Public License for more details.                      *
 *                                                                         *
 *   You should have received a copy of the GNU Library General Public     *
 *   License along with this library; if not, write to the Free Software   *
 *   Foundation, Inc., 59 Temple Place - Suite 330, Boston,                *
 *   MA 02111-1307, USA                                                    *
 ***************************************************************************/

#include <cstdio>

#include <cstdlib>

#include <boost/filesystem.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/algorithm/string.hpp>

#include "bocat/common/ini.hpp"

#include "store.hpp"

#include "cstore.hpp"


namespace	sakcab	{

#define	N_CHECKPOINT_INTERVAL	10000

///////////////////////////////////////////////////////////////////////
//  class KyotoStore
//

KyotoStore::KyotoStore( const char* ini_filename, const char* ini_section )
			:	_config( ini_filename, ini_section )
				
{
	// create cabs
	for (unsigned i=0; i<_config.cabs.size(); ++i) {
		const std::string&	cab_section = _config.cabs[ i ];
		cab_ptr	cab_p = new cab_type( ini_filename, cab_section.c_str() );
		_cabs.push_back( cab_p );
	}

	// create table cabs
	for (unsigned i=0; i<_config.table_cabs.size(); ++i) {
		const std::string&	cab_section = _config.table_cabs[ i ];

		bocat::common::INIFile	ini( ini_filename );
		const std::string&	SECTION = cab_section;
		std::string	table_name = ini[ SECTION ][ "table.name" ];
		if (table_name.size() < 1) {
			std::ostringstream	msg;
			msg << "No 'table.name' configured in [" << cab_section << "] in " << ini_filename;

			throw std::runtime_error( msg.str() );
		}
		if ( _table_cabs.count( table_name ) > 0 ) {
			std::ostringstream	msg;
			msg << "duplicate table name '" << table_name << "' configured in [" << cab_section << "] in " << ini_filename;
			throw std::runtime_error( msg.str() );
		}

		cab_ptr	cab_p = new cab_type( ini_filename, cab_section.c_str() );
		_table_cabs.insert( std::make_pair(table_name, cab_p) );
	}
}

KyotoStore::~KyotoStore() 
{
	for (unsigned i=0; i<_cabs.size(); ++i) {
		cab_ptr	cab_p = _cabs[ i ];
		delete cab_p;
	}
	_cabs.clear();

	table_cabs_type::iterator	it  = _table_cabs.begin();
	table_cabs_type::iterator	end = _table_cabs.end();
	while ( it != end ) {
		cab_ptr	cab_p = it->second;
		delete cab_p;

		++it;
	}
	_table_cabs.clear();
} 


///////////////////////////////////////////////////////////////////////
//


int     
KyotoStore::insert( const std::string& kpin,
                     const std::string& key, 
                     const std::string& data )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( cid );

    return  cab_p->insert( cid, kpin, key, data );
}


int     
KyotoStore::insertAppend( const std::string& kpin,
                     const std::string& key, 
                     const std::string& data )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( cid );

    return  cab_p->insertAppend( cid, kpin, key, data );
}


int     
KyotoStore::insertIgnore( const std::string& kpin,
                     const std::string& key, 
                     const std::string& data )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( cid );

    return  cab_p->insertIgnore( cid, kpin, key, data );
}


int     
KyotoStore::find( const std::string& kpin,
                   const std::string& key, 
                         std::string& buf  )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( cid );

    return  cab_p->find( cid, kpin, key, buf );
}


int     
KyotoStore::find( const std::string& kpin,
                   const std::string& key, 
                         std::vector< std::string >& all_data,
						 u_int32_t  limit )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( cid );

    return  cab_p->find( cid, kpin, key, all_data, limit );
}


int     
KyotoStore::erase( const std::string& kpin,
                    const std::string& key )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( cid );

    return  cab_p->erase( cid, kpin, key );
}


///////////////////////////////////////////////////////////////////////
//

int     
KyotoStore::insertAppend( const std::vector< std::string >& kpin_v,
						   const std::vector< std::string >& key_v,    
						   const std::vector< std::string >& data_v ) 
{
	int	count_failure = 0;

	for (unsigned i=0; i < kpin_v.size(); ++i) {
		const std::string&	kpin = kpin_v[ i ];
		const std::string&	key  = key_v[ i ];
		const std::string&	data = data_v[ i ];

		int cid = getCID( kpin );
		cab_ptr	cab_p = getCabPtr( cid );

		int	rc = cab_p->insertAppend( cid, kpin, key, data );
		if ( rc < 0 ) {
			++count_failure;
		}
	}

    return  -count_failure;
}


///////////////////////////////////////////////////////////////////////
//


int     
KyotoStore::insert( const std::string& table,
                    const std::string& kpin, 
                    const std::string& key, 
                    const std::string& data )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( table );
	if ( NULL == cab_p ) {
		std::ostringstream	msg;
		msg << "Table '" << table << "' is NOT configured";

		throw std::runtime_error( msg.str() );
	}

    return  cab_p->insert( cid, kpin, key, data );
}


int     
KyotoStore::insertAppend( const std::string& table,
                    	  const std::string& kpin, 
                     	  const std::string& key, 
                     	  const std::string& data )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( table );
	if ( NULL == cab_p ) {
		std::ostringstream	msg;
		msg << "Table '" << table << "' is NOT configured";

		throw std::runtime_error( msg.str() );
	}

    return  cab_p->insertAppend( cid, kpin, key, data );
}


int     
KyotoStore::insertIgnore( const std::string& table,
                    	  const std::string& kpin, 
						  const std::string& key, 
						  const std::string& data )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( table );
	if ( NULL == cab_p ) {
		std::ostringstream	msg;
		msg << "Table '" << table << "' is NOT configured";

		throw std::runtime_error( msg.str() );
	}

    return  cab_p->insertIgnore( cid, kpin, key, data );
}


int     
KyotoStore::find( const std::string& table,
                  const std::string& kpin, 
                  const std::string& key, 
                        std::string& buf  )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( table );
	if ( NULL == cab_p ) {
		std::ostringstream	msg;
		msg << "Table '" << table << "' is NOT configured";

		throw std::runtime_error( msg.str() );
	}

    return  cab_p->find( cid, kpin, key, buf );
}


int     
KyotoStore::find( const std::string& table,
                  const std::string& kpin, 
                  const std::string& key, 
                        std::vector< std::string >& all_data,
					    u_int32_t  limit )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( table );
	if ( NULL == cab_p ) {
		std::ostringstream	msg;
		msg << "Table '" << table << "' is NOT configured";

		throw std::runtime_error( msg.str() );
	}

    return  cab_p->find( cid, kpin, key, all_data, limit );
}


int     
KyotoStore::erase( const std::string& table,
                   const std::string& kpin, 
                   const std::string& key )
{
    int cid = getCID( kpin );
	cab_ptr	cab_p = getCabPtr( table );
	if ( NULL == cab_p ) {
		std::ostringstream	msg;
		msg << "Table '" << table << "' is NOT configured";

		throw std::runtime_error( msg.str() );
	}

    return  cab_p->erase( cid, kpin, key );
}


///////////////////////////////////////////////////////////////////////
//

int     
KyotoStore::insertAppend( const std::vector< std::string >& table_v,
						  const std::vector< std::string >& kpin_v,    
						  const std::vector< std::string >& key_v,    
						  const std::vector< std::string >& data_v ) 
{
	int	count_failure = 0;

	for (unsigned i=0; i < kpin_v.size(); ++i) {
		const std::string&	table = table_v[ i ];
		const std::string&	kpin = kpin_v[ i ];
		const std::string&	key  = key_v[ i ];
		const std::string&	data = data_v[ i ];

		cab_ptr	cab_p = getCabPtr( table );
		if ( NULL == cab_p ) {
			std::ostringstream	msg;
			msg << "Table '" << table << "' is NOT configured";

			throw std::runtime_error( msg.str() );
		}

		int cid = getCID( kpin );

		int	rc = cab_p->insertAppend( cid, kpin, key, data );
		if ( rc < 0 ) {
			++count_failure;
		}
	}

    return  -count_failure;
}


///////////////////////////////////////////////////////////////////////
//

int
KyotoStore::slim( void )
{
	int	count_record_slimmed = 0;
	for (unsigned i=0; i<_cabs.size(); ++i) {
		cab_ptr	cab_p = _cabs[ i ];

		int	rc = cab_p->slim();
		if ( rc < 0 ) {
			// failed
			continue;
		}

		count_record_slimmed += rc;
	}

	table_cabs_type::iterator	it  = _table_cabs.begin();
	table_cabs_type::iterator	end = _table_cabs.end();
	for ( ; it != end; ++it ) {
		cab_ptr	cab_p = it->second;

		int	rc = cab_p->slim();
		if ( rc < 0 ) {
			// failed
			continue;
		}

		count_record_slimmed += rc;
	}

	return	count_record_slimmed;
}



///////////////////////////////////////////////////////////////////////
//

KyotoStore::cid_type
KyotoStore::getCID( const std::string& kpin )
{
	// assert kpin.size() == 16 
	if ( 16 != kpin.size() ) {
		std::ostringstream	msg;
		msg << "BAD KPIN (size=" << kpin.size() << ", expect 16)";
		throw std::runtime_error( msg.str() );
	}

	const char*    p = kpin.c_str();
	u_int32_t	x = (p[0] << 24) + (p[1] << 16) + (p[2] << 8) + p[3];

	return	x;
}


KyotoStore::cab_ptr
KyotoStore::getCabPtr( cid_type cid )
{
	if (_cabs.size() < 1) {
		std::ostringstream	msg;
		msg << "No cabs available";
		throw std::runtime_error( msg.str() );
	}
	if (_cabs.size() == 1) {
		return	_cabs[0];
	}

	int	i = cid % _cabs.size();
	return	_cabs[ i ];
}


KyotoStore::cab_ptr
KyotoStore::getCabPtr( const std::string& table )
{
	table_cabs_type::iterator	it = _table_cabs.find( table );
	if ( _table_cabs.end() == it ) {
		return	NULL;
	}

	return	it->second;
}


///////////////////////////////////////////////////////////////////////
//

KyotoConfig::KyotoConfig( const char* ini_filename, const char* ini_section )
{
	this->parseConfig( ini_filename, ini_section );
}


int
KyotoConfig::parseConfig( const char* ini_filename, const char* ini_section )
{
	/*
 	 *       INI config
	 *   [sakcab::store]
	 *   envhome = /path/to/env/home
	 *   dbhome  = /path/to/db/home
	 *   datahome  = /path/to/data/home
	 *
	 */

	bocat::common::INIFile	ini( ini_filename );
	
    const char* SECTION = ini_section;

	// cabs
	{
		std::string	cab_str = ini[ SECTION ][ "store.cabs" ];
		if (cab_str.size() > 0) {
			std::vector< std::string >	v;
			boost::split( v, cab_str, boost::is_any_of(",") );
			for ( unsigned i=0; i<v.size(); ++i ) {
				std::string	cab = v[i];
				boost::trim( cab );
				cabs.push_back( cab );
			}
		}
	}

	// table.cabs
	{
		std::string	tcab_str = ini[ SECTION ][ "store.table.cabs" ];

		if (tcab_str.size() > 0) {
			std::vector< std::string >	v;
			boost::split( v, tcab_str, boost::is_any_of(",") );
			for ( unsigned i=0; i<v.size(); ++i ) {
				std::string	cab = v[i];
				boost::trim( cab );
				table_cabs.push_back( cab );
			}
		}
	}

	if ( (cabs.size() + table_cabs.size()) < 1 ) {
		char	msg[512];
		snprintf(msg, sizeof(msg) - 1, "No 'store.table.cabs' or 'store.cabs' configured in [%s] of file %s", SECTION, ini_filename);
		throw std::runtime_error( msg );
	}

	return 0;
}

}	//// namespace sakcab

